package com.shengzai.flink.core

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, PreparedStatement}

object Demo4ClazzNum {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val stuDS: DataStream[String] = env.readTextFile("data/students.txt")

    val stuInfoDS: DataStream[Students] = stuDS
      .map(line => {
        val split: Array[String] = line.split(",")
        val id: String = split(0)
        val name: String = split(1)
        val age: Int = split(2).toInt
        val sex: String = split(3)
        val clazz: String = split(4)
        new Students(id,name,age,sex,clazz)
      })
    stuInfoDS.addSink(new ToMysql)

    env.execute()

  }

}
class ToMysql extends RichSinkFunction[Students]{
  var conn :Connection=_
  var stat: PreparedStatement=_

  override def open(parameters: Configuration): Unit = {
    Class.forName("com.mysql.jdbc.Driver")
    conn = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata22?useUnicode=true&characterEncoding=UTF-8", "root", "123456")
    stat = conn.prepareStatement("insert into student2 values(?,?,?,?,?)")

  }

  override def close(): Unit = {
    conn.close()
  }

  override def invoke(value: Students , context: SinkFunction.Context): Unit = {
    stat.setString(1,value.id)
    stat.setString(2,value.name)
    stat.setInt(3,value.age)
    stat.setString(4,value.sex)
    stat.setString(5,value.clazz)
    stat.execute()
  }
}
case class Students(id:String,name:String,age:Int,sex:String,clazz:String)